package sfs

import (
	"context"
	"crypto/sha1"
	"encoding/hex"
	"fmt"
	"github.com/chicken-team-outside/chicken_transmission/api_messages"
	"github.com/chicken-team-outside/chicken_transmission/driver"
	"github.com/chicken-team-outside/chicken_transmission/socket"
	"github.com/chicken-team-outside/chicken_transmission/utils"
	"github.com/golang/protobuf/proto"
	"go.uber.org/zap"
	"io"
	"net/http"
	"path/filepath"
	"slices"
	"sync"
	"sync/atomic"
	"time"
)

const MaxMessageSize = 8 * 1024 * 1024
const DefaultDirectSendMaxSize = 100 * 1024

var ErrNotSetNetDiskDriver = fmt.Errorf("not set net disk driver")
var ErrNotSetCacheDir = fmt.Errorf("not set net disk cache dir")
var ErrNotSetLocalDriver = fmt.Errorf("not set local driver")
var ErrChecksumNotEqual = fmt.Errorf("checksum not equal")

type FsClient struct {
	rpcClient           *socket.WebSocketClient
	local               atomic.Value
	netDisk             atomic.Value
	netDiskCacheDir     atomic.Value
	timeout             atomic.Int64
	httpClient          atomic.Value
	directSendMaxSize   atomic.Int64
	uploadBlockLock     *utils.NamedLock
	createUploadDirLock sync.Mutex
}

func (p *FsClient) SetNetDiskCacheDir(cacheDir string) {
	p.netDiskCacheDir.Store(&cacheDir)
}

func (p *FsClient) getNetDiskCacheDir() *string {
	val := p.netDiskCacheDir.Load()
	if val == nil {
		return nil
	}
	return val.(*string)
}

func (p *FsClient) SetLocal(local driver.LocalDriver) {
	p.local.Store(&atomicValue{
		Value: local,
	})
}

func (p *FsClient) SetNetDisk(disk driver.NetDiskDriver) {
	p.netDisk.Store(&atomicValue{
		Value: disk,
	})
}

func (p *FsClient) GetLocal() driver.LocalDriver {
	val := p.local.Load().(*atomicValue).Value
	if val == nil {
		return nil
	}
	return val.(driver.LocalDriver)
}

func (p *FsClient) GetNetDisk() driver.NetDiskDriver {
	val := p.netDisk.Load().(*atomicValue).Value
	if val == nil {
		return nil
	}
	return val.(driver.NetDiskDriver)
}

func (p *FsClient) SetHttpClient(client *http.Client) {
	p.httpClient.Store(client)
}

func (p *FsClient) getHttpClient() *http.Client {
	val := p.httpClient.Load()
	if val == nil {
		return http.DefaultClient
	}
	return val.(*http.Client)
}

func (p *FsClient) sendRequest(ctx context.Context, clientId uint64, t api_messages.MessageType, body proto.Message) ([]byte, error) {
	reply, err := p.send(ctx, clientId, 0, t, body, true)
	if err != nil {
		return nil, err
	}
	defer reply.Close()
	timeout, cancelFunc := context.WithTimeout(ctx, time.Duration(p.timeout.Load())*time.Second)
	defer cancelFunc()
	select {
	case rp := <-reply.ReplyChan:
		return rp.Body, nil
	case <-timeout.Done():
		return nil, socket.ErrTimeout
	}
}

func (p *FsClient) sendReplyByHeader(ctx context.Context, header *api_messages.MessageHeader, body proto.Message) error {
	_, err := p.send(ctx, header.ClientId, header.Id, header.Type, body, false)
	return err
}

func (p *FsClient) getCurConContext() context.Context {
	return p.rpcClient.GetCurConContext()
}

func (p *FsClient) onMessage(ctx context.Context, header *api_messages.MessageHeader, body []byte) {
	var resolver socket.ClientOnMessageFunc
	if header.Reply == 0 {
		switch header.Type {
		case api_messages.MessageType_ListDir:
			resolver = p.listDirProvider
		case api_messages.MessageType_ReadFile:
			resolver = p.readFileProvider
		case api_messages.MessageType_WriteFile:
			resolver = p.writeFileProvider
		case api_messages.MessageType_RemoveAll:
			resolver = p.removeAllProvider
		case api_messages.MessageType_MkdirAll:
			resolver = p.mkdirProvider
		case api_messages.MessageType_Stat:
			resolver = p.statProvider
		case api_messages.MessageType_Rename:
			resolver = p.renameProvider
		case api_messages.MessageType_Truncate:
			resolver = p.TruncateProvider
		case api_messages.MessageType_CalculateFileBlocksSha1:
			resolver = p.calculateFileBlocksSha1Provider
		}
		if resolver != nil {
			go resolver(ctx, header, body)
		} else {
			zap.L().Warn("not found provider", zap.Any("header", header))
		}
	}
}

func (p *FsClient) SetTimeout(timeout int64) {
	if timeout > 0 {
		p.timeout.Store(timeout)
	}
}

func (p *FsClient) createUploadDir(ctx context.Context, dirPath string) error {
	disk := p.GetNetDisk()
	if disk == nil {
		return ErrNotSetNetDiskDriver
	}
	p.createUploadDirLock.Lock()
	defer p.createUploadDirLock.Unlock()
	return disk.MkdirAll(ctx, filepath.ToSlash(dirPath))
}

func (p *FsClient) uploadBlock(ctx context.Context, data io.ReadSeeker, size int64, sha1CheckSum []byte) (*api_messages.LinkInfo, []byte, error) {
	disk := p.GetNetDisk()
	if disk == nil {
		return nil, nil, ErrNotSetNetDiskDriver
	}
	cacheDir := p.getNetDiskCacheDir()
	if cacheDir == nil {
		return nil, nil, ErrNotSetCacheDir
	}
	fileSha1Bytes, err := utils.Sha1ForReader(data, nil)
	if err != nil {
		return nil, nil, err
	}
	if len(sha1CheckSum) == sha1.Size && !slices.Equal(sha1CheckSum, fileSha1Bytes) {
		return nil, nil, ErrChecksumNotEqual
	}
	fileSha1Str := hex.EncodeToString(fileSha1Bytes)
	_, err = data.Seek(0, io.SeekStart)
	if err != nil {
		return nil, nil, err
	}
	parentPath := filepath.Join(*cacheDir, fileSha1Str[:2])
	err = p.createUploadDir(ctx, parentPath)
	if err != nil {
		return nil, nil, err
	}
	resultPath := filepath.Join(parentPath, fileSha1Str[2:]+".exe")
	unlockFunc := p.uploadBlockLock.Lock(resultPath)
	defer unlockFunc()
	exist, err := disk.Exists(ctx, resultPath)
	if err == nil && !exist {
		err = disk.UploadFile(ctx, size, resultPath, data)
	}
	if err != nil {
		return nil, nil, err
	}
	link, err := disk.GetFileDownloadLink(ctx, resultPath)
	return link, fileSha1Bytes, err
}

func (p *FsClient) downloadBlock(ctx context.Context, offset, end int64, link *api_messages.LinkInfo) (io.ReadCloser, error) {
	request, err := http.NewRequestWithContext(ctx, "GET", link.Link, nil)
	if err != nil {
		return nil, err
	}
	for k, v := range link.Headers {
		request.Header.Set(k, v)
	}
	request.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", offset, end))
	var rsp *http.Response
	rsp, err = p.getHttpClient().Do(request)
	if err != nil {
		return nil, err
	}
	if rsp.StatusCode != http.StatusPartialContent && rsp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("invalid status code(%d): %s", rsp.StatusCode, ReadHttpErrorMsg(rsp.Body))
	}
	return rsp.Body, nil
}

func (p *FsClient) send(ctx context.Context, clientId, replay uint64, t api_messages.MessageType, content proto.Message, receiveMsg bool) (reply *socket.ReplayContext, err error) {
	return p.rpcClient.SendProtobuf(ctx, p.timeout.Load(), clientId, replay, t, content, receiveMsg)
}

func (p *FsClient) SetDirectSendMaxSize(size int64) {
	p.directSendMaxSize.Store(size)
}

func (p *FsClient) Disconnect() {
	p.rpcClient.Disconnect()
}

func (p *FsClient) IsConnected() bool {
	return p.rpcClient.IsConnected()
}

func (p *FsClient) Connect(ctx context.Context, address string, token string) (err error) {
	return p.rpcClient.Connect(ctx, address, http.Header{
		"token": {token},
	})
}

func NewFsProvider(onConnectClose func(), statusChange socket.ClientStatusChangeHandler) (p *FsClient) {
	p = &FsClient{}
	p.local.Store(new(atomicValue))
	p.netDisk.Store(new(atomicValue))
	p.rpcClient = socket.NewWebSocketClient(MaxMessageSize, p.onMessage, statusChange, onConnectClose)
	p.timeout.Store(10)
	p.directSendMaxSize.Store(DefaultDirectSendMaxSize)
	p.uploadBlockLock = utils.NewNamedLock()
	return
}
